package com.deleyQueue;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import lombok.Data;
import org.springframework.util.CollectionUtils;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.Tuple;

import java.lang.reflect.Type;
import java.util.Calendar;
import java.util.Set;
import java.util.UUID;

/**
 * @author tiny
 * @date 2022/2/11 15:02
 * @Description:
 */
public class RedisDelayingQueue<T> {
    @Data
    class TaskItem<T>{
        String id;
        T msg;
    }

    private Type TaskType = new TypeReference<TaskItem<T>>(){}.getType();

    private Jedis jedis;
    private String queueKey;

    public RedisDelayingQueue(Jedis jedis,String queueKey){
        this.jedis = jedis;
        this.queueKey = queueKey;
    }

    /**
     * 序列化后指定延长时长放入zset队列
     * @return
     */
    public boolean delay(T msg,long delaySeconds){
        TaskItem<T> task = new TaskItem<>();
        task.setId(UUID.randomUUID().toString());
        task.setMsg(msg);
        jedis.zadd(queueKey, System.currentTimeMillis()+delaySeconds*1000, JSON.toJSONString(task));
        return true;
    }

    public void loop(){
        while( !Thread.interrupted() ){
            //只查询一条
            Set<Tuple> values = jedis.zrangeByScoreWithScores(queueKey, 0, System.currentTimeMillis(), 0, 1);
            //一条到期的也没有时就sleep一会
            if( CollectionUtils.isEmpty(values) ){
                try{
                    Thread.sleep(500);
                }catch (InterruptedException ex){
                    break;
                }
                continue;
            }
            Tuple tuple = values.iterator().next();
            //可以优化的点：用lua脚本让zrangeByScoreWithScores查找+zrem删除变成原子操作
            if(tuple.getScore() >= (Calendar.getInstance().getTimeInMillis())/1000 && jedis.zrem(queueKey,tuple.getElement())>0){
                TaskItem<T> taskItem = JSON.parseObject(tuple.getElement(), TaskType);
                this.handleMsg(taskItem.getMsg());
            }
        }

    }

    private void handleMsg(T msg) {
        System.out.println("处理延时任务"+msg);
    }

}
